package com.zatech.cgnci.project.base.utils;

import com.google.common.collect.Lists;

import java.util.List;
import java.util.concurrent.*;
import java.util.function.Function;

/**
 * 线程池工具类.
 *
 * @author user
 */
public class ThreadPoolUtils {

    private static volatile ThreadPoolExecutor threadPool;
    private static volatile ScheduledThreadPoolExecutor scheduledThreadPool;

    public static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 6 + 1;
    public static final int MAX_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 8 + 1;

    public static final int KEEP_ALIVE_TIME = 1000;
    public static final int BLOCK_QUEUE_SIZE = 3000;

    public static void executor(Runnable runnable) {
        getThreadPoolExecutor().execute(runnable);
    }

    public static void schedule(Runnable runnable, int delay, TimeUnit unit) {
        getScheduledThreadPool().schedule(runnable, delay, unit);
    }

    public static void scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
        getScheduledThreadPool().scheduleWithFixedDelay(command, initialDelay, delay, unit);
    }

    public static <T> Future<T> submit(Callable<T> callable) {
        return getThreadPoolExecutor().submit(callable);
    }

    /**
     * 获取线程池对象
     *
     * @return 线程池对象
     */
    public static ThreadPoolExecutor getThreadPoolExecutor() {
        if (threadPool == null) {
            synchronized (ThreadPoolUtils.class) {
                if (threadPool == null) {
                    threadPool = new ThreadPoolExecutor(
                            CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, TimeUnit.SECONDS,
                            new LinkedBlockingQueue<>(BLOCK_QUEUE_SIZE),
                            new ThreadPoolExecutor.CallerRunsPolicy()
                    );
                }
            }
        }
        return threadPool;
    }

    /**
     * 获取一个定时的线程池.
     *
     * @return 定时线程池对象
     */
    public static ScheduledThreadPoolExecutor getScheduledThreadPool() {
        if (scheduledThreadPool == null) {
            synchronized (ThreadPoolUtils.class) {
                if (scheduledThreadPool == null) {
                    scheduledThreadPool = new ScheduledThreadPoolExecutor(
                            CORE_POOL_SIZE,
                            Executors.defaultThreadFactory(),
                            new ThreadPoolExecutor.CallerRunsPolicy()
                    );
                }
            }
        }
        return scheduledThreadPool;
    }


    /**
     * 批量执行，并且返回是否成功的标志
     *
     * @param dataList       数据列表
     * @param partitionCount 并行执行的数量
     * @param function       Function
     * @param <T>            数据类型
     * @return 批量执行是否成功
     */
    public static <T> boolean batchExecute(List<T> dataList, Integer partitionCount, Function<T, Boolean> function) {
        List<List<T>> partition = Lists.partition(dataList, partitionCount);
        return partition.stream().map(list -> list.stream()
                .map(data -> CompletableFuture.supplyAsync(() -> function.apply(data), getThreadPoolExecutor()))
                .map((future) -> {
                    try {
                        return future.get();
                    } catch (InterruptedException | ExecutionException e) {
                        e.printStackTrace();
                        return Boolean.FALSE;
                    }
                }).filter(flag -> flag).count() == list.size()).filter(flag -> flag).count() == partition.size();
    }
}